// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: leon

#include "coin2/exchange/phemex_base/feed_v1/processor.h"

#include <set>
#include <string_view>

#include "coin2/base/conversion.h"
#include "coin2/base/string_util.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/phemex_base/feed_v1/update.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::phemex_base::feed_v1::impl {
namespace {
using boost::gregorian::date;
using boost::posix_time::ptime;
using boost::posix_time::time_duration;
static const date epoch = date(1970, 1, 1);

int64_t CalculateFundingTime(int64_t ts) {
  ptime cur = PtimeFromTimestamp(ts);
  ptime utc8 = ptime(cur.date(), time_duration(8, 0, 0));
  ptime utc16 = ptime(cur.date(), time_duration(16, 0, 0));
  ptime utc24 = ptime(cur.date(), time_duration(24, 0, 0));
  if (cur < utc8) {
    return (utc8 - ptime(epoch)).total_nanoseconds();
  } else if (cur < utc16) {
    return (utc16 - ptime(epoch)).total_nanoseconds();
  } else if (cur < utc24) {
    return (utc24 - ptime(epoch)).total_nanoseconds();
  } else {
    NOTREACHED();
  }
}

void ParseTicker(
    const rapidjson::Value& ticker_json,
    int64_t ts,
    base::feed::impl::FeedUpdateManager* upd_mgr,
    double price_multiplier) {
  const std::string& native_symbol = ticker_json["symbol"].GetString();
  if (!upd_mgr->is_symbol_registered(native_symbol)) {
    return;
  }
  if (ticker_json.HasMember("openInterest")) {
    auto* open_interest = upd_mgr->GetOpenInterestMutable();
    open_interest->timestamp = ts;
    open_interest->open_interest_qty = stoll(ticker_json["openInterest"].GetString());
    upd_mgr->PublishOpenInterestWithExchangeTimeCheck(native_symbol);
  }
  if (ticker_json.HasMember("markPrice")) {
    auto* mark_price = upd_mgr->GetMarkPriceMutable();
    mark_price->timestamp = ts;
    mark_price->mark_price = stoll(ticker_json["markPrice"].GetString()) * price_multiplier;
    upd_mgr->PublishMarkPriceWithExchangeTimeCheck(native_symbol);
  }
  if (ticker_json.HasMember("indexPrice")) {
    auto* index = upd_mgr->GetIndexMutable();
    index->timestamp = ts;
    index->price = stoll(ticker_json["indexPrice"].GetString()) * price_multiplier;
    upd_mgr->PublishIndexWithExchangeTimeCheck(native_symbol);
  }
  if (ticker_json.HasMember("fundingRate") &&
      ticker_json.HasMember("predFundingRate")) {
    auto* funding_rate = upd_mgr->GetFundingRateMutable();
    funding_rate->timestamp = ts;
    funding_rate->funding_rate = stoll(ticker_json["fundingRate"].GetString()) * 1e-8;
    funding_rate->estimated_rate = stoll(ticker_json["predFundingRate"].GetString()) * 1e-8;
    funding_rate->funding_time = CalculateFundingTime(ts);
    upd_mgr->PublishFundingRateWithExchangeTimeCheck(native_symbol);
  }
}

// "fields": [
//   "symbol",
//   "openRp",
//   "highRp",
//   "lowRp",
//   "lastRp",
//   "volumeRq",
//   "turnoverRv",
//   "openInterestRv",
//   "indexRp",
//   "markRp",
//   "fundingRateRr",
//   "predFundingRateRr"
// ],
void ParseHedgedTicker(
    const rapidjson::Value& ticker_json,
    int64_t ts,
    base::feed::impl::FeedUpdateManager* upd_mgr,
    double price_multiplier) {
  for (const auto& data : ticker_json.GetArray()) {
    const auto& items = data.GetArray();
    CHECK_EQ(items.Size(), 12);
    const std::string& native_symbol = items[0].GetString();
    if (!upd_mgr->is_symbol_registered(native_symbol)) {
      continue;;
    }

    auto* open_interest = upd_mgr->GetOpenInterestMutable();
    open_interest->timestamp = ts;
    open_interest->open_interest_qty = stod(items[7].GetString());
    upd_mgr->PublishOpenInterestWithExchangeTimeCheck(native_symbol);

    auto* mark_price = upd_mgr->GetMarkPriceMutable();
    mark_price->timestamp = ts;
    mark_price->mark_price = stod(items[9].GetString()) * price_multiplier;
    upd_mgr->PublishMarkPriceWithExchangeTimeCheck(native_symbol);

    auto* index = upd_mgr->GetIndexMutable();
    index->timestamp = ts;
    index->price = stod(items[8].GetString()) * price_multiplier;
    upd_mgr->PublishIndexWithExchangeTimeCheck(native_symbol);


    auto* funding_rate = upd_mgr->GetFundingRateMutable();
    funding_rate->timestamp = ts;
    funding_rate->funding_rate = stod(items[10].GetString());
    funding_rate->estimated_rate = stod(items[11].GetString());
    funding_rate->funding_time = CalculateFundingTime(ts);
    upd_mgr->PublishFundingRateWithExchangeTimeCheck(native_symbol);

  }
}

}  // namespace

double PhemexBaseFeedProcessor::GetPriceMultiplier(std::string_view native_symbol) {
  if(upd_mgr_.mea().market == coin::proto::Spot) {
    // https://api.phemex.com/public/products
    // thier priceScale is 4
    static const char* quotes[] = {
      "TRY", "BRZ"
    };
    for (std::string_view q: quotes) {
      if (StringEndsWith(native_symbol, q)) {
        return 1e-4;
      }
    }
  }

  return price_multiplier_;
}

void PhemexBaseFeedProcessor::OnTopicRecord(const TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  const auto& data = packet.payload;

  rapidjson::Document document;

  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(data.c_str());
  if (document.HasParseError()) {
    THROW() << "JSON parse error: "
            << GetParseError_En(document.GetParseError()) << packet.payload;
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  const auto channel = ExtractChannel(document);
  if (!CheckChannelFilter(channel)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }
  if (document.HasMember("symbol")) {
    const std::string& native_symbol = document["symbol"].GetString();
    double price_multiplier = GetPriceMultiplier(native_symbol);
    if (document.HasMember("book") || document.HasMember("orderbook_p")) {
      const std::string& type = document["type"].GetString();
      int64_t ts = stoll(document["timestamp"].GetString());
      auto* book = upd_mgr_.GetBookMutable(native_symbol);
      if (book == nullptr) {
        upd_mgr_.onEndPacket(packet);
        return;
      }

      const rapidjson::Value* book_json = nullptr;
      if (document.HasMember("book")) {
        book_json = &(document["book"]);
      } else if (document.HasMember("orderbook_p")) {
        book_json = &(document["orderbook_p"]);
      }
      bool generated = false;
      if (type == "snapshot") {
        generated = PhemexBaseFeedUpdateParser::ParseSnapshot(
            book, ts, book_json, price_multiplier, qty_multiplier_);
      } else if (type == "incremental") {
        generated = PhemexBaseFeedUpdateParser::ParseIncremental(
            book, ts, book_json, price_multiplier, qty_multiplier_);
      } else {
        THROW() << "[Phemex Feed Processor] unkown book type: " << packet.payload;
      }
      if (generated) {
        upd_mgr_.PublishBook();
      }
    } else if (document.HasMember("trades") || document.HasMember("trades_p")) {
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        upd_mgr_.onEndPacket(packet);
        return;
      }
      const std::string& type = document["type"].GetString();
      if (type == "snapshot") {
        upd_mgr_.onEndPacket(packet);
        return;
      }

      const rapidjson::Value* trade_json = nullptr;
      if (document.HasMember("trades")) {
        trade_json = &(document["trades"]);
      } else if (document.HasMember("trades_p")) {
        trade_json = &(document["trades_p"]);
      }
      Trade* agg_trade = upd_mgr_.GetTradeMutable();
      agg_trade->Clear();
      Trade trade{};

      for (auto& trade_json : trade_json->GetArray()) {
        PhemexBaseFeedUpdateParser::ParseTrade(
            trade_json, &trade, price_multiplier, qty_multiplier_);
        if (TryAggregateTrades(agg_trade, trade)) {
          continue;
        }
        if (agg_trade->fill_qty > 0) {
          upd_mgr_.PublishTrade(native_symbol, true);
        }
        *agg_trade = trade;
      }
      if (agg_trade->fill_qty > 0) {
        upd_mgr_.PublishTrade(native_symbol, false);
      }
    } else {
      THROW() << "[Phemex Feed Processor] unkown topic: " << packet.payload;
    }
  } else if (document.HasMember("market24h")) {
    int64_t ts = stoll(document["timestamp"].GetString());
    ParseTicker(document["market24h"], ts, &upd_mgr_, price_multiplier_);
  } else if (channel == "perp_market24h_pack_p") {
    int64_t ts = stoll(document["timestamp"].GetString());
    ParseHedgedTicker(document["data"], ts, &upd_mgr_, price_multiplier_);
  } else if (channel == "kline_rest") {
    auto url = document["url"].GetString();
    std::vector<std::string> splited = Split(url, "?");
    CHECK_EQ(splited.size(), 2);
    auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
    const std::string& native_symbol = parameters["symbol"];
    auto interval = PhemexBaseFeedUpdateParser::ParseKlineInterval(parameters["resolution"]);
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines != nullptr) {
      const auto& data = document["data"]["data"]["rows"].GetArray();
      std::optional<int64_t> first_open_timestamp;
      if (data.Size() > 0) {
        first_open_timestamp = stoll(data[data.Size() - 1][0].GetString()) * 1'000'000'000LL;
      }

      if (klines->Size() == 0) {
        klines->SnapshotClear();
      } else if (first_open_timestamp &&
                 first_open_timestamp < klines->RBegin()->open_timestamp) {
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      bool to_publish = false;
      double price_multiplier = GetPriceMultiplier(native_symbol);
      for (int i = data.Size() - 1; i >= 0; i--) {
        const auto& item = data[i];
        bool freezed = PhemexBaseFeedUpdateParser::Parse1Kline(
            item,
            klines,
            packet.timestamp,
            interval,
            price_multiplier,
            qty_multiplier_);
        to_publish = to_publish || freezed;
      }
      if (to_publish) upd_mgr_.PublishKline(native_symbol);
    }

    

  } else if (document.HasMember("result")) {
    LOG(INFO) << "[Phemex Feed Processor] meta msg for " << packet.topic_name << ": "
                << packet.payload.data();
  } else {
    THROW() << "[Phemex Feed Processor] unkown topic: " << packet.payload;
  }

  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::phemex_base::feed_v1::impl
